package com.tang.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.connector.base.DeliveryGuarantee;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.kafka.clients.producer.ProducerConfig;

/**
 * 输出到Kafka
 * 注意：如果要使用 精准一次 写入Kafka，需要满足以下条件，缺一不可
 * 1、开启checkpoint（后续介绍）
 * 2、设置事务前缀
 * 3、设置事务超时时间：   checkpoint间隔 <  事务超时时间  < max的15分钟
 * ========
 * 创建队列
 * bin/kafka-topics.sh --create --zookeeper 192.168.70.141:2181 --replication-factor 1 --partitions 1 --topic ws
 * 监听消息
 * bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic ws --from-beginning
 *
 * @author tang
 * @since 2023/6/9 16:43
 */
public class SinkKafka {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 如果是精准一次，必须开启checkpoint（后续章节介绍）
        env.enableCheckpointing(2000, CheckpointingMode.EXACTLY_ONCE);

        DataStreamSource<String> source = env.socketTextStream("192.168.70.141", 7777);

        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
                .setBootstrapServers("192.168.70.141:9092") // 多个需要使用逗号分割开 ip:port,ip:port......
                .setRecordSerializer(
                        KafkaRecordSerializationSchema.<String>builder()
                                .setTopic("ws") // 输出topic名称
                                .setValueSerializationSchema(new SimpleStringSchema())
                                .build()
                )
                // 写入到kafka的一致性级：精准一次、至少一次
                .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                // 如果是精准一次，必须设置 事务的前缀
                .setTransactionalIdPrefix("ws-")
                // 如果是精准一次，必须设置 事务超时时间: 大于checkpoint间隔，小于 max 15分钟
                .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG, 10 * 60 * 1000 + "")
                .build();

        source.sinkTo(kafkaSink);

        env.execute();
    }

}
